package de.tuberlin.dima.aim3.oc.input.jaxb;

import java.io.Reader;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;

import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBElement;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Unmarshaller;
import javax.xml.stream.EventFilter;
import javax.xml.stream.XMLEventReader;
import javax.xml.stream.XMLInputFactory;
import javax.xml.stream.XMLStreamException;
import javax.xml.stream.events.XMLEvent;

import org.mediawiki.xml.export_0.CommentType;
import org.mediawiki.xml.export_0.ContributorType;
import org.mediawiki.xml.export_0.PageType;
import org.mediawiki.xml.export_0.RevisionType;
import org.mediawiki.xml.export_0.TextType;

import de.tuberlin.dima.aim3.oc.Constants;
import de.tuberlin.dima.aim3.oc.input.IterativeWikiDumpParser;
import de.tuberlin.dima.aim3.oc.input.WikiDumpParser;
import de.tuberlin.dima.aim3.oc.input.custom.WikiPage;
import de.tuberlin.dima.aim3.oc.input.custom.WikiRevision;
import de.tuberlin.dima.aim3.oc.input.custom.WikiUser;
import eu.stratosphere.pact.common.io.FileInputFormat;

/**
 * Parser for Wikipedia XML dumps based on JAXB and StAX which iteratively reads
 * and parses an input file and returns Java instances for every successfully
 * parsed node.
 * 
 * This parser is optimized for stratosphere's {@link FileInputFormat} API and
 * therefore allows simple repetitive parsing of XML to Java instances. It holds
 * an internal pointer to the XML file and returns the next Java instance for
 * every call to {@link #readNextWikiDumpPage()}. It furthermore supports the
 * {@link FileInputFormat#reachedEnd()} method by checking its internal pointer.
 * 
 * @author Florian Feigenbutz <florian.feigenbutz@campus.tu-berlin.de>
 * 
 */
public class WikiDumpStaxJaxbParser extends WikiDumpParser implements
    IterativeWikiDumpParser {

  @SuppressWarnings("rawtypes")
  private static final Class PARSED_CLASS = PageType.class;

  private final Reader xmlReader;

  private XMLEventReader staxReader;

  private XMLEventReader staxFiltRd;

  private JAXBContext jaxbContext;

  private Unmarshaller unmarshaller;

  private final NodeConverter nodeHandler;

  // ----------------------------------- //

  public WikiDumpStaxJaxbParser(Reader xmlReader) {
    this.xmlReader = xmlReader;
    nodeHandler = new PageConverterHandler();

    try {
      init();
    } catch (Throwable e) {
      throw new RuntimeException("Error while initializing Wiki Dump parser", e);
    }
  }

  protected void init() throws XMLStreamException, JAXBException {

    // StAX:
    EventFilter startElementFilter = new EventFilter() {
      @Override
      public boolean accept(XMLEvent event) {
        return event.isStartElement();
      }
    };

    XMLInputFactory staxFactory = XMLInputFactory.newInstance();
    staxReader = staxFactory.createXMLEventReader(xmlReader);
    staxFiltRd = staxFactory.createFilteredReader(staxReader,
        startElementFilter);

    // skip StartElement:
    staxFiltRd.nextEvent();

    // JAXB:
    jaxbContext = JAXBContext.newInstance(PARSED_CLASS);
    unmarshaller = jaxbContext.createUnmarshaller();
  }

  @Override
  public WikiPage readWikiDumpPage(byte[] pageAsBytes) {
    // TODO Auto-generated method stub
    return null;
  }

  // ----------- IterativeWikiDumpParser API ---------- //

  @Override
  public WikiPage readNextWikiDumpPage() {
    return readNextWikiDumpPage(0);
  }

  private WikiPage readNextWikiDumpPage(int numberOfRetries) {
    try {
      if (staxFiltRd.peek() != null) {
        Object node = unmarshaller.unmarshal(staxReader, PARSED_CLASS);
        if (node instanceof JAXBElement
            && ((JAXBElement<?>) node).getValue() != null) {
          node = ((JAXBElement<?>) node).getValue();
        }
        WikiPage wikiPage = (WikiPage) nodeHandler.convert(node);
        // the first element might be null so give it a single retry
        if (wikiPage == null && numberOfRetries < 1) {
          numberOfRetries++;
          wikiPage = readNextWikiDumpPage(numberOfRetries);
        }
        return wikiPage;
      }
    } catch (XMLStreamException e) {
      // FIXME Auto-generated catch block
      e.printStackTrace();
    } catch (JAXBException e) {
      // FIXME Auto-generated catch block
      e.printStackTrace();
    }
    return null;
  }

  @Override
  public boolean reachedEnd() {
    return !staxFiltRd.hasNext();
  }

  // ----------- WikiDumpParser API ---------- //

  @Override
  public List<WikiPage> readWikiDumpPage() throws Exception {

    List<WikiPage> results = new ArrayList<WikiPage>();
    // Parsing:
    while (staxFiltRd.peek() != null) {
      Object node = unmarshaller.unmarshal(staxReader, PARSED_CLASS);
      if (node instanceof JAXBElement
          && ((JAXBElement<?>) node).getValue() != null) {
        node = ((JAXBElement<?>) node).getValue();
      }
      WikiPage wikiPage = (WikiPage) nodeHandler.convert(node);
      if (wikiPage != null) {
        results.add(wikiPage);
      }
    }
    return results;
  }

  // ----------------------------------------------------------- //

  private interface NodeConverter {
    Object convert(Object node);
  }

  private class PageConverterHandler implements NodeConverter {

    private final DateFormat dateFormatter;

    public PageConverterHandler() {
      super();
      dateFormatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZZ",
          Constants.LOCALE);
    }

    @Override
    public Object convert(Object node) {
      if (!(node instanceof PageType)) {
        throw new IllegalArgumentException(
            "Node to be converted must be of type " + PageType.class.getName()
                + " - was " + node.getClass().getName());
      }

      PageType inputPage = (PageType) node;

      // skip empty results which happen e.g. during errornous xml input
      if (inputPage.getId() == null || inputPage.getTitle() == null) {
        return null;
      }

      List<WikiRevision> outputRevisions = new ArrayList<WikiRevision>(
          inputPage.getRevisionOrUploadOrLogitem().size());
      for (Object revisionOrUploadOrLogitem : inputPage
          .getRevisionOrUploadOrLogitem()) {
        if (revisionOrUploadOrLogitem instanceof RevisionType) {
          RevisionType inputRevision = (RevisionType) revisionOrUploadOrLogitem;

          ContributorType inputUser = inputRevision.getContributor();
          String userId = getStringValue(inputUser.getId());
          WikiUser outputUser = new WikiUser(inputUser.getUsername(), userId,
              inputUser.getIp());

          boolean isMinorChange = inputRevision.getMinor() != null;
          String formattedTimestamp = dateFormatter.format(inputRevision
              .getTimestamp().toGregorianCalendar().getTime());
          String comment = getStringValue(inputRevision.getComment());
          String text = getStringValue(inputRevision.getText());
          WikiRevision outputRevision = new WikiRevision(inputRevision.getId()
              .toString(), formattedTimestamp, outputUser, isMinorChange,
              comment, text);
          outputRevisions.add(outputRevision);
        }
      }

      WikiPage outputPage = new WikiPage(inputPage.getTitle(), inputPage
          .getId().toString(), outputRevisions);
      return outputPage;
    }

    private String getStringValue(Object input) {
      String outputValue;
      if (input == null) {
        outputValue = null;
      } else if (input instanceof CommentType) {
        outputValue = ((CommentType) input).getValue().trim();
      } else if (input instanceof TextType) {
        outputValue = ((TextType) input).getValue().trim();
      } else {
        outputValue = input.toString();
      }
      return input != null ? outputValue : null;
    }
  }

}
